package com.atguigu.gmall.realtime.util;

import com.atguigu.gmall.realtime.common.Constant;

/**
 * @Author lzc
 * @Date 2023/2/13 09:59
 */
public class SQLUtil {
    
    public static String getDDLKafkaSource(String topic, String groupId, String... format) {
        String defaultFormat = "json";// 没有传入数据格式, 使用默认格式
        if (format.length != 0) {  // 有传入
            defaultFormat = format[0];
        }
        
        return "with(" +
            "'connector' = 'kafka', " +
            "'properties.bootstrap.servers' = '" + Constant.KAFKA_BROKERS + "', " +
            "'properties.group.id' = '" + groupId + "', " +
            "'topic' = '" + topic + "', " +
            "'scan.startup.mode' = 'latest-offset', " +
            ("json".equals(defaultFormat) ? "'json.ignore-parse-errors' = 'true'," : "") +  // 当解析失败的时候, 忽略错误
            "'format' = '" + defaultFormat + "' " +
            ")";
    }
    
    public static String getDDLKafkaSink(String topic, String... format) {
        String defaultFormat = "json";// 没有传入数据格式, 使用默认格式
        if (format.length != 0) {  // 有传入
            defaultFormat = format[0];
        }
    
        return "with(" +
            "'connector' = 'kafka', " +
            "'sink.semantic' = 'exactly-once', " +
            "'properties.transaction.timeout.ms' = '900000', " +
            "'properties.bootstrap.servers' = '" + Constant.KAFKA_BROKERS + "', " +
            "'topic' = '" + topic + "', " +
            "'format' = '" + defaultFormat + "' " +
            ")";
    }
    
    public static String getDDLUpsertKafka(String topic, String ... format) {
        String defaultFormat = "json";// 没有传入数据格式, 使用默认格式
        if (format.length != 0) {  // 有传入
            defaultFormat = format[0];
        }
    
        return "with(" +
            "'connector' = 'upsert-kafka', " +
            "'properties.bootstrap.servers' = '" + Constant.KAFKA_BROKERS + "', " +
            "'topic' = '" + topic + "', " +
            "'key.format' = '" + defaultFormat + "', " +
            "'value.format' = '" + defaultFormat + "' " +
            ")";
    }
}
